using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Mutation;
using Org.Apache.Cassandra.Thrift;
using Thrift;

namespace Me.Prettyprint.Cassandra.Service.Template
{
    /// <summary>
    /// The wrapper enumerator makes it easier to 
    /// </summary>
    /// <typeparam name="DictionaryEntry"></typeparam>
    public class WrapperEnumerator
    {
        public WrapperEnumerator(ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> data)
        {
            Data = data;
            DataDictionary = data.Get;
            Keys = new List<byte[]>(DataDictionary.Keys);
            Position = -1;
        }
        protected List<byte[]> Keys
        {
            get;
            set;
        }
        protected int Position
        {
            get;
            set;
        }
        protected ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> Data
        {
            get;
            set;
        }
        protected Dictionary<byte[], List<ColumnOrSuperColumn>> DataDictionary
        {
            get;
            set;
        }
        public byte[] Current
        {
            get
            {
                return Keys[Position];
            }
        }
        public byte[] Next()
        {
            Position++;
            return Keys[Position];
        }
        /// <summary>
        /// HasNext checks to see if the list.Count is > Positio + 1
        /// </summary>
        /// <returns></returns>
        public bool HasNext()
        {
            return Keys.Count > (Position + 1);
        }
        /// <summary>
        /// Reset sets the position to -1, which indicates the wrapper hasn't started enumerating yet
        /// </summary>
        public void Reset()
        {
            Position = -1;
        }
    }
    /// <summary>
    ///  Wraps the results with as an Iterator. The underlying Iterator has already been advanced
    ///  to the first row upon construction.
    /// </summary>
    /// <typeparam name="K"></typeparam>
    /// <typeparam name="N"></typeparam>
	public class ColumnFamilyResultWrapper<K,N> : AbstractResultWrapper<K,N>
	{
        public ColumnFamilyResultWrapper(Serializer<K> keySerializer,
            Serializer<N> columnNameSerializer,
            ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> executionResult)
            : base(keySerializer, columnNameSerializer, executionResult)
        {
            ExecutionResult = executionResult;
            ResultRows = executionResult.Get;
            Enumerator = new WrapperEnumerator(executionResult);
            HasEntries = Enumerator.HasNext();
            HasResults = HasEntries;
        }
        private WrapperEnumerator Enumerator
        {
            get;
            set;
        }
        internal ExecutionResult<Dictionary<byte[], List<ColumnOrSuperColumn>>> ExecutionResult
        {
            get;
            set;
        }
        protected Dictionary<N, HColumn<N, byte[]>> Columns
        {
            get;
            set;
        }
        protected Dictionary<byte[], List<ColumnOrSuperColumn>> ResultRows
        {
            get;
            set;
        }
        protected bool HasEntries
        {
            get;
            set;
        }
        private byte[] CurrentKey
        {
            get;
            set;
        }
        public override List<N> GetColumnNames()
        {
            if (Columns == null)
            {
                return new List<N>();
            }
            else
            {
                return new List<N>(Columns.Keys);
            }
        }
        public override ByteBuffer GetColumnValue(N columnName)
        {
            HColumn<N, byte[]> col = GetColumn(columnName);
            return col != null ? new ByteBuffer(col.ValueBytes) : null;
        }
        public override HColumn<N, byte[]> GetColumn(N columnName)
        {
            return Columns[columnName];
        }
        public K GetKey()
        {
            return KeySerializer.FromBytes(CurrentKey);
        }
        private void ApplyToRow(List<ColumnOrSuperColumn> cosclist)
        {
            HasEntries = cosclist.Count > 0;
            for (IEnumerator<ColumnOrSuperColumn> iterator = cosclist.GetEnumerator(); iterator.MoveNext(); )
            {
                ColumnOrSuperColumn cosc = iterator.Current;
                if (cosc.__isset.super_column)
                {
                    ApplySuper(cosc);
                }
                else
                {
                    ApplyStandard(cosc.Column);
                }
            }
        }
        private void ApplySuper(ColumnOrSuperColumn cosc)
        {
            IEnumerator<Column> tcolumns = cosc.Super_column.Columns.GetEnumerator();
            while (tcolumns.MoveNext())
            {
                ApplyStandard(tcolumns.Current);
            }
        }
        private void ApplyStandard(Column cosc)
        {
            N colName = ColumnNameSerializer.FromBytes(cosc.Name);
            HColumn<N, byte[]> column = null;
            // Dictionary works differently than Java HashMap, so we do a try otherwise it will throw an exception. Fix submitted by Valentin P
            if (!Columns.TryGetValue(colName, out column))
            {
                column = new HColumnImpl<N, byte[]>(cosc, ColumnNameSerializer, BytesArraySerializer.Instance);
            }
            else
            {
                ((HColumnImpl<N, ByteBuffer>)column).Apply(cosc);
            }
            Columns[colName] = column;
        }
        public override ColumnFamilyResult<K, N> Next()
        {
            if (!HasNext())
            {
                throw new NullReferenceException("No more rows left on this HColumnFamily");
            }
            CurrentKey = Enumerator.Next();
            Key = KeySerializer.FromBytes(CurrentKey);
            Columns = new Dictionary<N, HColumn<N, byte[]>>();
            List<ColumnOrSuperColumn> colSupercol = ResultRows[CurrentKey];
            ApplyToRow(colSupercol);
            return this;
        }
        public override bool HasNext()
        {
            return Enumerator.HasNext();
        }
        public void Reset()
        {
            Enumerator = null;
            Enumerator = new WrapperEnumerator(ExecutionResult);
        }
	}
}